package com.gma.userprofile.util

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client.{ConnectionFactory, Put, Table}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.sql.DataFrame

object HBaseUtil {

  /**
   * 将用户表标签导入hbase
   * @param tag:数据
   * @param tagName：标签名称
   */
  def tagToHBase(tag: DataFrame, tagName: String): Unit = {

    //4、将结果保存到hbase中
    tag.foreachPartition(iter => {

      //创建hbase数据库链接（启动hbase,导入依赖）
      val conf = new Configuration()
      conf.set("hbase.zookeeper.quorum", "master")
      val con = ConnectionFactory.createConnection(conf)
      //获取表对象(需要先在hbase中创建表)
      //create 'userprofile','info'
      val table: Table = con.getTable(TableName.valueOf("userprofile"))

      //循环分区内的数据，将数据保存到hbase中
      iter.foreach(row => {
        val id = row.getAs[Int]("id")

        val sex: Object = row.getAs[Object](tagName)

        //构建put对象，以用户编号作为rowkey
        val put = new Put(Bytes.toBytes(id))
        //增加列
        put.addColumn("info".getBytes, tagName.getBytes(), sex.toString.getBytes)
        //插入数据
        table.put(put)
      })

      //关闭数据库链接
      con.close()
    })


  }


}
